package com.shujia.hw;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.sql.*;
import java.util.Properties;

public class Mysql2Hbase {
    public static void main(String[] args) throws IOException, ClassNotFoundException, SQLException {
        // 读取Mysql中的数据
        // 添加配置文件
        Properties properties = new Properties();
        properties.load(new FileReader("src/main/resources/jdbc.properties"));

        // 注册驱动
        Class.forName(properties.getProperty("driver"));

        // 创建连接
        Connection connection = DriverManager.getConnection(
                properties.getProperty("url"),
                properties.getProperty("user"),
                properties.getProperty("password")
        );


        // 获取操作数据对象
        PreparedStatement preparedStatement = connection.prepareStatement("select * from student");


        /**
         * 创建HBase连接
         */

        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum","node1,node2,master");
        org.apache.hadoop.hbase.client.Connection hbCon = ConnectionFactory.createConnection(conf);

        Table table = hbCon.getTable(TableName.valueOf("api:tb1"));



        // 获取数据
        ResultSet resultSet = preparedStatement.executeQuery();
        while (resultSet.next()){
            String id = resultSet.getString("id");
            String name = resultSet.getString("name");
            int age = resultSet.getInt("age");
            String gender = resultSet.getString("gender");
            String clazz = resultSet.getString("clazz");

            Put put = new Put(Bytes.toBytes(id));

            /**
             * public KeyValue(final byte [] row, final byte [] family,
             *       final byte [] qualifier, final byte [] value)
             */
            KeyValue nameKeyValue = new KeyValue(
                    Bytes.toBytes(id),
                    Bytes.toBytes("info1"),
                    Bytes.toBytes("name"),
                    Bytes.toBytes(name)
            );

            put.add(nameKeyValue);

            KeyValue ageKeyValue = new KeyValue(
                    Bytes.toBytes(id),
                    Bytes.toBytes("info1"),
                    Bytes.toBytes("age"),
                    Bytes.toBytes(age)
            );

            put.add(ageKeyValue);

            KeyValue genderKeyValue = new KeyValue(
                    Bytes.toBytes(id),
                    Bytes.toBytes("info1"),
                    Bytes.toBytes("gender"),
                    Bytes.toBytes(gender)
            );

            put.add(genderKeyValue);


            KeyValue clazzKeyValue = new KeyValue(
                    Bytes.toBytes(id),
                    Bytes.toBytes("info2"),
                    Bytes.toBytes("clazz"),
                    Bytes.toBytes(clazz)
            );

            put.add(clazzKeyValue);
            table.put(put);
        }

        table.close();
        hbCon.close();

        preparedStatement.close();
        connection.close();





    }
}
